package com.aiwiown.snackmq.network.client;

import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.network.config.NetworkConfig;
import com.aiwiown.snackmq.network.handler.ClientHandler;
import com.aiwiown.snackmq.network.handler.ClientMessageDispatcher;
import com.aiwiown.snackmq.network.handler.HeartbeatHandler;
import com.aiwiown.snackmq.network.handler.MessageCodec;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

import java.io.Closeable;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

@Slf4j
public class NettyClient implements Closeable {

    private final NetworkConfig networkConfig;
    private final Bootstrap bootstrap;
    private final EventLoopGroup group;
    private Channel channel;
    private final ClientHandler clientHandler = new ClientHandler();
    private final ClientMessageDispatcher clientMessageDispatcher = new ClientMessageDispatcher();


    /**
     * 构造函数，初始化 Netty 客户端。
     *
     * @param networkConfig 网络配置对象，包含连接所需的主机、端口、超时等信息。
     */
    public NettyClient(NetworkConfig networkConfig) {
        this.networkConfig = networkConfig;
        this.group = new NioEventLoopGroup();
        this.bootstrap = new Bootstrap();

        // 配置 Netty 客户端的引导程序
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, networkConfig.getConnectionTimeout())
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline pipeline = ch.pipeline();
                        // 添加空闲状态处理器，用于心跳检测
                        pipeline.addLast(new IdleStateHandler(0, networkConfig.getHeartbeatInterval(), 0, TimeUnit.MILLISECONDS));
                        // 添加基于长度字段的帧解码器，用于处理消息分帧
                        pipeline.addLast(new LengthFieldBasedFrameDecoder(2 * 1024 * 1024, 5, 4, 0, 0));
                        // 添加消息编解码器，用于序列化和反序列化消息
                        pipeline.addLast(new MessageCodec());
                        // 添加心跳处理器，用于发送和接收心跳包
                        pipeline.addLast(new HeartbeatHandler(true));
                        // --- 【新增】将客户端分发器加入 pipeline，使其能处理入站请求 ---
                        pipeline.addLast(clientMessageDispatcher);
                        // 添加客户端处理器，用于处理业务逻辑
                        pipeline.addLast(clientHandler);
                    }
                });
    }

    /**
     * 连接到服务器。
     *
     * @throws InterruptedException 如果连接过程中线程被中断
     */
    public void connect() throws InterruptedException {
        log.info("Connecting to server {}:{}", networkConfig.getHost(), networkConfig.getPort());
        ChannelFuture future = bootstrap.connect(networkConfig.getHost(), networkConfig.getPort()).sync();
        this.channel = future.channel();
        log.info("Connected to server {}/{}", channel.remoteAddress(), channel.localAddress());
    }

    /**
     * 重连到新的服务器地址。
     *
     * @param newAddress 新的 Leader 地址，格式为 "host:port"
     * @throws InterruptedException 如果连接过程中线程被中断
     * @throws IllegalArgumentException 如果地址格式无效或端口无法解析
     */
    /**
     * 重连到新的服务器地址。
     *
     * @param newAddress 新的 Leader 地址，格式为 "host:port"
     * @throws InterruptedException 如果连接过程中线程被中断
     * @throws IllegalArgumentException 如果地址格式无效或端口无法解析
     * @throws RuntimeException 如果连接超时或连接失败
     */
    public void reconnect(String newAddress) throws InterruptedException, IllegalArgumentException, RuntimeException {
        if (newAddress == null || newAddress.trim().isEmpty()) {
            log.warn("Attempted to reconnect with an invalid (null or empty) address.");
            throw new IllegalArgumentException("New address cannot be null or empty");
        }
        log.info("Reconnecting to new address: {}", newAddress);
        // 关闭当前连接（如果存在）
        if (this.channel != null && this.channel.isActive()) {
            this.channel.close().awaitUninterruptibly();
        }
        // 解析新地址
        String[] parts = newAddress.split(":");
        if (parts.length != 2) {
            log.error("Invalid redirect address format: {}. Expected host:port.", newAddress);
            throw new IllegalArgumentException("Invalid redirect address format: " + newAddress);
        }

        String host = parts[0];
        int port;
        try {
            port = Integer.parseInt(parts[1]);
        } catch (NumberFormatException e) {
            log.error("Invalid port in redirect address: {}", newAddress);
            throw new IllegalArgumentException("Invalid port in redirect address: " + newAddress, e);
        }

        // 更新网络配置以反映新的地址
        this.networkConfig.setHost(host);
        this.networkConfig.setPort(port);

        // 使用新的地址重新连接
        ChannelFuture future = bootstrap.connect(host, port).sync();
        this.channel = future.channel();
        log.info("Reconnected successfully to {}/{}", channel.remoteAddress(), channel.localAddress());
    }

    /**
     * 关闭客户端并释放资源。
     */
    public void shutdown() {
        log.info("Shutting down client...");
        if (channel != null) {
            channel.close();
        }
        group.shutdownGracefully();
        log.info("Client shutdown completed");
    }

    /**
     * 实现 Closeable 接口，关闭客户端。
     *
     * @throws IOException 如果关闭过程中发生 I/O 错误
     */
    @Override
    public void close() throws IOException {
        shutdown();
    }

    /**
     * 发送请求消息并返回异步结果。
     *
     * @param message 请求消息对象
     * @return 包含响应消息的 CompletableFuture 对象
     */
    public CompletableFuture<Message> sendRequest(Message message) {
        if (channel == null || !channel.isActive()) {
            CompletableFuture<Message> future = new CompletableFuture<>();
            future.completeExceptionally(new IllegalStateException("Client is not connected"));
            return future;
        }
        return clientHandler.sendRequest(message, channel);
    }

    /**
     * 检查客户端是否处于活动状态。
     *
     * @return 如果客户端已连接且通道处于活动状态，则返回 true；否则返回 false
     */
    public boolean isActive() {
        return this.channel != null && this.channel.isActive();
    }

    /**
     * 获取当前客户端配置的远程地址。
     *
     * @return 格式为 "host:port" 的地址字符串，如果未配置则返回 null
     */
    public String getRemoteAddress() {
        if (networkConfig != null) {
            return networkConfig.getHost() + ":" + networkConfig.getPort();
        }
        return null;
    }

    /**
     * 【新增】获取客户端消息分发器，以便注册处理器。
     * @return ClientMessageDispatcher 实例
     */
    public ClientMessageDispatcher getDispatcher() {
        return this.clientMessageDispatcher;
    }
}
